package com.webull.information.center.carwler.job.branch;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.webull.information.center.carwler.common.model.BranchNews;
import com.webull.information.center.carwler.common.model.CarwlerJobConfig;
import com.webull.information.center.carwler.common.util.CarwlerJobStateUtil;
import com.webull.information.center.carwler.common.util.eljob.ElasticJobUtil;
import com.webull.information.center.carwler.dao.CarwlerJobConfigDao;
import com.webull.information.center.carwler.dao.CarwlerNewsDao;
import com.webull.information.center.carwler.task.support.TikcerNewsProcessor;

/**
 * 处理各分支同步过来的news数据
 */
@Component
public class Base_Process_Branch_Job implements SimpleJob {
    protected final Logger logger = LogManager.getLogger(getClass());
    private @Autowired
    CarwlerJobStateUtil carwlerJobStateUtil;
    private @Autowired
    CarwlerNewsDao carwlerNewsDao;
    private @Autowired
    CarwlerJobConfigDao carwlerJobConfigDao;

    private @Autowired
    TikcerNewsProcessor tikcerNewsProcessor;
    private final String collectJob = "base_Process_Branch_Job";

	@Override
	public void execute(ShardingContext context) {
		int sharding = Optional.ofNullable(context).map(v -> v.getShardingItem()).orElse(-1);
		int totalCount = Optional.ofNullable(context).map(v -> v.getShardingTotalCount()).orElse(-1);
		carwlerJobStateUtil.assertCarwlerJobConfig(true);

		logger.info("Base_Process_Branch_Job start:" + ArrayUtils.toString(sharding) + ",totalCount:" + totalCount);

		List<CarwlerJobConfig> carwlerJobConfigs = Optional.ofNullable(carwlerJobConfigDao.searchBranchCarwler())
				.orElse(new ArrayList<>());
		List<CarwlerJobConfig> shardRegions = Optional
				.ofNullable(ElasticJobUtil.shardingItems(context, carwlerJobConfigs))
				.orElse(carwlerJobConfigs);
		for (CarwlerJobConfig tgs : shardRegions) {
			processBranchNews(tgs);
		}
	}

    private void processBranchNews(CarwlerJobConfig carwlerJobConfig) {
        logger.info("Base_Process_Branch_Job.processBranchNews div size{};carwlerJobConfig:{};", carwlerJobConfig);
        CarwlerJobConfig config = new CarwlerJobConfig();
        BeanUtils.copyProperties(carwlerJobConfig, config, "pastNewsId", "pastBaRelId");
        try {

            logger.info("Base_Process_Branch_Job.processBranchNews start collectJob{} div ,carwlerJobConfig:{} ",
                    collectJob, config);

            Integer lastId_ = Optional.ofNullable(carwlerJobConfigDao.getPastNewsIdByCarwler(config.getCarwler())).orElse(0);
            do1:
            while (true) {
                List<BranchNews> branchNews = carwlerNewsDao.searchBranchNews0(config.getEdenNewsTable(), lastId_);
                if (branchNews == null || branchNews.isEmpty()) break do1;

                branchNews.forEach(v -> {
                    try {
                        tikcerNewsProcessor.bindBranchNews(v);
                    } catch (Exception e) {
                        logger.error("Base_Process_Branch_Job error", e);
                    }
                });

                lastId_ = branchNews.get(branchNews.size() - 1).getId().intValue();
                config.setPastNewsId(lastId_);
                carwlerJobConfigDao.update(config);
                logger.info("Base_Process_Branch_Job.processBranchNews mid collectJob{} div ,carwlerJobConfig:{},lastId_{}",
                        collectJob, config.getCarwler(), lastId_);
            }
            logger.info("Base_Process_Branch_Job.processBranchNews end collectJob{} div ,carwlerJobConfig:{},lastId_{}",
                    collectJob, config.getCarwler(), lastId_);
        } catch (Exception e) {
            logger.error("Base_Process_Branch_Job.processBranchNews collectJob div error:" + config, e);
        }
    }


}
